
                /*********************************************
                **                                          **
                **                                          **
                **                 ca.cmr.C                 **
                **                                          **
                **                                          **
                *********************************************/
                /*********************************************
                **                                          **
                **         Author Jean-Marc Adamo           **
                **       Copyright (C) 1997-2008 JMA        **
                *********************************************/


/*
** needs even number of processes
*/

#include "arch.h"
#include "A.utilities"

//user program

Global_id gid1;
Coll_id cid1, cid2, cid3, cid4, cid5, cid6, cid7;

//on most homogenous systems, explicit setting 
//won't be necessary (try commenting the
//setting lines, you should get the same values
//for the cids).
void init(int, char **){
   gid1.set();

   cid1.set();
   cid2.set();
   cid3.set();
   cid4.set();
   cid5.set();
   cid6.set();
   cid7.set();
   fprintf(Sdf::trace, "node%d: gid1=%d, cid1=%d, cid2=%d, cid3=%d, cid4=%d, cid5=%d, cid6=%d, cid7=%d\n",
           Sdf::PROC_ID, 
           gid1.get(), cid1.get(), cid2.get(), cid3.get(),
           cid4.get(), cid5.get(), cid6.get(), cid7.get());
   fflush(Sdf::trace);
}

class SR_count{
   public:
      int cnt;
      SR_count(){
         cnt = 0;
      }
      void reset(){
         cnt = 0;
      }
      void more_SR(int i){
         cnt += i;
      }
};

#define wait_while(condition) while(condition) Sdf::poll();
#define not_complete  (c.cnt >0)

ChanA<int> *chan;

void upon_end_of_SR_r(ChanAs *chan){

   fprintf(Sdf::trace, "node%d, r>>>>>>>\n", Sdf::PROC_ID);
   Sdf::Access_type access_type;
   int tag_dv = chan->get_tag_dv(&access_type);
   fprintf(Sdf::trace, "node%d, upon_end_of_SR_r, capa=%d, rcvd size=%d, Tsize=%d, rps=%d, rpr=%d, tag_dv=%d, access_type=%d, tag_av=%d, tag_ev=%d, comm=%d, comm_type=%d\n", 
                Sdf::PROC_ID, 
                chan->get_chan_capacity(), 
                chan->get_msg_size(Sdf::RCV), 
                chan->get_Tsize(),
                chan->get_remote_proc(Sdf::SND), 
                chan->get_remote_proc(Sdf::RCV),
                tag_dv,
                access_type,
                chan->get_tag_av(),
                chan->get_tag_ev(),
                chan->get_comm(),
                chan->get_comm()->get_Comm_type());

   ((SR_count *)(chan->get_hndlr_data_ptr(Sdf::RCV)))->cnt--;

   fprintf(Sdf::trace, "node%d, r<<<<<<<\n", Sdf::PROC_ID);
   fflush(Sdf::trace);
}

void upon_end_of_SR_s(ChanAs *chan){           

   fprintf(Sdf::trace, "node%d, s>>>>>>>\n", Sdf::PROC_ID);
   Sdf::Access_type access_type;
   int tag_dv = chan->get_tag_dv(&access_type);
   fprintf(Sdf::trace, "node%d, upon_end_of_SR_s, capa=%d, sent size=%d, Tsize=%d, rps=%d, rpr=%d, access_type=%d, tag_dv=%d, tag_av=%d, tag_ev=%d, comm=%d, comm_type=%d\n", 
                Sdf::PROC_ID, 
                chan->get_chan_capacity(), 
                chan->get_msg_size(Sdf::SND), 
                chan->get_Tsize(),
                chan->get_remote_proc(Sdf::SND), 
                chan->get_remote_proc(Sdf::RCV),
                tag_dv,
                access_type,
                chan->get_tag_av(),
                chan->get_tag_ev(),
                chan->get_comm(),
                chan->get_comm()->get_Comm_type());

   ((SR_count *)(chan->get_hndlr_data_ptr(Sdf::SND)))->cnt--;

   fprintf(Sdf::trace, "node%d, s<<<<<<<\n", Sdf::PROC_ID);
   fflush(Sdf::trace);
}


class Root_proc: public Thread{

   public:

        Root_proc()
                : Thread((Tmf0_ptr)&Root_proc::body){}

        void body(){

                fprintf(Sdf::trace, "node %d, in Root_proc::start\n", Sdf::PROC_ID);
                fflush(Sdf::trace); 

                const int N = Sdf::PROC_NB/2;
                if(Sdf::PROC_NB%2){
                   printf("node %d, !!!!!execute with even process number: terminated\n", Sdf::PROC_ID);
                   fflush(stdout);
                   return;
                }

                Group *group1;
                int *rk1 = new int[N];
                int i;
                for(i = 0; i < N; i++)
                   rk1[i] = i;
                group1 = Sdf::incl(Sdf::COMML_WORLD, N, rk1);
                CommL *comm1 = group1->commit(cid1, Sdf::COMML_WORLD);

                Group *group2;
                int *rk2 = new int[N];
                for(i = N; i < Sdf::PROC_NB; i++)
                   rk2[i-N] = i;
                group2 = Sdf::incl(Sdf::COMML_WORLD, Sdf::PROC_NB - N, rk2);
                CommL *comm2 = group2->commit(cid2, Sdf::COMML_WORLD);

                Sdf::barrier(cid4);

                CommR *commr;
                commr = Sdf::newCommR(cid3, comm1, comm2, Sdf::COMML_WORLD);

                  //each process in local group communicates with same 
                  //rank process in remote group.
                int rank = commr->get_local_CommL()->rank();
                chan = new ChanA<int>(gid1, rank, 1, commr);

                SR_count c;
                chan->set(Sdf::SND, upon_end_of_SR_s, &c);
                chan->set(Sdf::RCV, upon_end_of_SR_r, &c);

                Sdf::barrier(cid5);
                  //synchronizes chan construction over Sdf::COMML_WORLD(no longer
                  //necessary starting from version 5)

                int from = Sdf::PROC_ID;
                int to = 777;

                c.more_SR(2);
                if(Sdf::PROC_ID < N){
                   chan->send(Sdf::STD, from);
                   chan->recv(to);
                }else{
                   chan->recv(to);
                   chan->send(Sdf::STD, from);
                }
                wait_while(not_complete);
                Sdf::barrier(cid6);

                fprintf(Sdf::trace, "node %d, from=%d, to=%d\n", 
                        Sdf::PROC_ID, from, to);
                fflush(Sdf::trace);

                delete []rk1;
                delete []rk2;

                Sdf::barrier(cid7);
                     //synchronizes channel destruction
                delete comm1;
                delete comm2;
                delete commr;
                delete chan;

                fprintf(Sdf::trace, "node %d, the end\n", Sdf::PROC_ID);
                fflush(Sdf::trace);
        }
};
//end of user program

MAIN(Root_proc, init, Sdf::empty, Sdf::empty)

